package com.surfilter.massdata.spark.util.ipcheck;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.storage.StorageLevel;

import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.surfilter.massdata.spark.bean.ipcheck.ErrIp;
import com.surfilter.massdata.spark.bean.ipcheck.IpBean;
import com.surfilter.massdata.spark.bean.ipcheck.ObjectInfo;

import scala.Tuple2;
import scala.Tuple3;

/**
 * @author hutao
 * @date 创建时间：2017年5月24日 下午5:06:49
 * @Description：来源冲突比对、分配冲突比对、扫描来源库，参照分配库、扫描分配库，参照来源库
 * @version 1.0
 * 
 **/
public class IpCheckAnalyz {
	private static final Log log = LogFactory.getLog(IpCheckAnalyz.class);
	public static  JavaRDD<IpBean> dealBusiness(JavaRDD<IpBean> ipBeanRdd,SQLContext sc) throws Exception {
		JavaSparkContext jsc = JavaSparkContext.fromSparkContext(sc.sparkContext());
		final Broadcast<Map<Long, String>> broadBswdMap = jsc.broadcast(IpCheckUtil.getBsdwMap());//广播运营商ID
		final Broadcast<Map<String, String>> broadOptorIds = jsc.broadcast(IpCheckUtil.getOptorIdsMap());//广播本省单位
		JavaPairRDD<Long, Iterable<IpBean>> secondRDD = IpCheckUtil.transJavaPariRdd(ipBeanRdd);
		
		 JavaRDD<Tuple3<Object, Boolean, Long>> newRdd=secondRDD.flatMap(new FlatMapFunction<Tuple2<Long,Iterable<IpBean>>, Tuple3<Object,Boolean,Long>>() {

			private static final long serialVersionUID = 1L;

			@Override
			public Iterable<Tuple3<Object, Boolean, Long>> call(Tuple2<Long, Iterable<IpBean>> tuple) throws Exception {
				// TODO Auto-generated method stub
				long ip = tuple._1;
				Iterable<IpBean> its = tuple._2;
				Set<String> excludeSameSet = Sets.newHashSet();
				List<IpBean> reduceSourceList = Lists.newArrayList();
				List<IpBean> reduceAllocateList = Lists.newArrayList();
				for (IpBean ipBean : its) {
					boolean flag = true;
					// 这里做排重
					flag = excludeSame(ipBean, excludeSameSet);
					if (flag) {
						if (ipBean.getTflag().equals(IpCheckConstants.IP_SOURCE_TYPE)) {
							reduceSourceList.add(ipBean);
						} else if (ipBean.getTflag().equals(IpCheckConstants.IP_ALLOCATE_TYPE)) {
							reduceAllocateList.add(ipBean);
						}
					}
				}
				List<Tuple3<Object,Boolean,Long>> resultList=new ArrayList<Tuple3<Object,Boolean,Long>>();
				ObjectInfo objectInfo =null;
				if(reduceSourceList.size()>0){
					objectInfo = IpSourceConflictChecker.sourceConflictCheck(reduceSourceList);
				}
				if(reduceAllocateList.size()>0 && ((null==objectInfo)||objectInfo.isResult()==true)){
					objectInfo=IpAllocateConflictChecker.allocateConflictCheck(reduceAllocateList,broadOptorIds.getValue());
				}
				if(reduceSourceList.size()>0&& ((null==objectInfo)||objectInfo.isResult()==true)){
					objectInfo=IpSourceStorageScanner.scanSourceStorage(reduceSourceList, reduceAllocateList, broadBswdMap.getValue());
				}
				if(reduceAllocateList.size()>0&& ((null==objectInfo)||objectInfo.isResult()==true)){
					objectInfo=IpAllocateStorageScanner.scanAllocateStorage(reduceSourceList, reduceAllocateList, broadBswdMap.getValue());
				}
				if(null!=objectInfo){
					boolean flag=objectInfo.isResult();
					if(flag){
						//正确数据
						reduceSourceList.addAll(reduceAllocateList);
						for(IpBean bean:reduceSourceList){
							resultList.add(new Tuple3<Object, Boolean, Long>(bean,flag,ip));
						}
					}else{
						ErrIp errIp=objectInfo.getErrInfo();
						resultList.add(new Tuple3<Object, Boolean, Long>(errIp,flag,ip));
					}
				}
			
				return resultList;
			}

		});
		 newRdd.persist(StorageLevel.MEMORY_AND_DISK_SER());
//		 ########################合并冲突数据并入库冲突表##########################
		 JavaPairRDD<String, Long> errRdd=newRdd.filter(new Function<Tuple3<Object,Boolean,Long>, Boolean>() {
				private static final long serialVersionUID = 1L;

				public Boolean call(Tuple3<Object, Boolean, Long> v1) throws Exception {
					Boolean flag=v1._2();
					boolean result=false;
					if(null!=flag&&flag==false){
						result=true;
					}
					return result;
				}
			}).mapToPair(new PairFunction<Tuple3<Object,Boolean,Long>, String, Long>() {

				private static final long serialVersionUID = 1L;

				@Override
				public Tuple2<String, Long> call(Tuple3<Object, Boolean, Long> tuple3) throws Exception {
					// TODO Auto-generated method stub
					ErrIp errIp=(ErrIp)tuple3._1();
					return new Tuple2<String,Long>(errIp.toString(),tuple3._3());
				}
			
			});
			 JavaRDD<ErrIp> conflictRdd= errRdd.groupByKey().flatMap(new FlatMapFunction<Tuple2<String,Iterable<Long>>, ErrIp>() {
				private static final long serialVersionUID = 1L;

				@Override
				public Iterable<ErrIp> call(Tuple2<String, Iterable<Long>> tuple2) throws Exception {
					String conflictStr=tuple2._1;
					Iterable<Long> its=tuple2._2;
					List<ErrIp> resultList=new ArrayList<ErrIp>();
					String[] array=conflictStr.split("\t");
					ErrIp errIp = new ErrIp();
					if(array.length==5){
						errIp = ConvertUtil.convertErrIp(array, errIp);
					}
					
					List<Long> list=new ArrayList<Long>();
					for(Long ip:its){
						list.add(ip);
					}
					List<String> segList=IpUtils.getIpSeg(list);
					for(String seg:segList){
						long qsip=Long.parseLong(seg.split(",")[0]);
						long zzip=Long.parseLong(seg.split(",")[1]);
						ErrIp source = ConvertUtil.copyErrIp(new ErrIp(), errIp);
						source.setQsip(qsip);
						source.setZzip(zzip);
						resultList.add(source);
					}
					return resultList;
				}
			});
			 DataFrame conlinctDF=sc.createDataFrame(conflictRdd, ErrIp.class);
			 IpCheckUtil.writeConflictToTable(conlinctDF);//将冲突数据入库
			 
//			 conflictRdd.foreach(new VoidFunction<ErrIp>() {
//					@Override
//					public void call(ErrIp t) throws Exception {
//						// TODO Auto-generated method stub
//						System.err.println("error>>>>>>"+t.getQsip()+"\t"+t.getZzip()+"\t"+t.getBbdws()+"\t"+t.getConflictType()+"\t"+t.getRegionIds()+"\t"+t.getStorageType()+"\t"+t.getJsonStr());
//					}
//				});
//		
//			 ########################合并冲突数据并入库冲突表结束##########################
			 
			 
			 
			 
//		 ################################合并无冲突数据开始#########################################################################
		 JavaPairRDD<String, Long> rightRdd=newRdd.filter(new Function<Tuple3<Object,Boolean,Long>, Boolean>() {
			private static final long serialVersionUID = 1L;

			public Boolean call(Tuple3<Object, Boolean, Long> v1) throws Exception {
				Boolean flag=v1._2();
				return null!=flag&&flag==true?true:false;
			}
		}).mapToPair(new PairFunction<Tuple3<Object,Boolean,Long>, String, Long>() {

			private static final long serialVersionUID = 1L;

			@Override
			public Tuple2<String, Long> call(Tuple3<Object, Boolean, Long> tuple3) throws Exception {
				// TODO Auto-generated method stub
				IpBean bean=(IpBean)tuple3._1();
				bean.setIsPass(null);
				return new Tuple2<String,Long>(bean.toNewString(),tuple3._3());
			}
		
		});
		 JavaRDD<IpBean> resultRdd= rightRdd.groupByKey().flatMap(new FlatMapFunction<Tuple2<String,Iterable<Long>>, IpBean>() {
			private static final long serialVersionUID = 1L;

			@Override
			public Iterable<IpBean> call(Tuple2<String, Iterable<Long>> tuple2) throws Exception {
				String beanStr=tuple2._1;
				Iterable<Long> its=tuple2._2;
				List<IpBean> resultList=new ArrayList<IpBean>();
				String[] array=beanStr.split("\t");
				IpBean ipBean = new IpBean();
				if(array.length==46){
					ipBean = ConvertUtil.convertIpBean(array, ipBean);
					List<Long> list=new ArrayList<Long>();
					for(Long ip:its){
						list.add(ip);
					}
					List<String> segList=IpUtils.getIpSeg(list);
					for(String seg:segList){
						long qsip=Long.parseLong(seg.split(",")[0]);
						long zzip=Long.parseLong(seg.split(",")[1]);
						IpBean source = ConvertUtil.copy(new IpBean(), ipBean);
						source.setQsip(qsip);
						source.setZzip(zzip);
						resultList.add(source);
					}
				}
			
				return resultList;
			}
		});
//		 ################################合并无冲突数据结束#########################################################################
//		resultRdd.foreach(new VoidFunction<IpBean>() {
//			
//			@Override
//			public void call(IpBean t) throws Exception {
//				// TODO Auto-generated method stub
//				System.err.println("right>>>>>>"+t.getBbdw_id()+"\t"+t.getLyid()+"\t"+t.getId()+"\t"+t.getQsip()+"\t"+t.getZzip());
//			}
//		});
	 
		return  resultRdd;
		
	}

	/**
	 * @Description:ip去重
	 * @param ipBean
	 * @param set
	 * @return
	 */
	public static boolean excludeSame(IpBean ipBean, Set<String> set) {
		boolean flag = true;
		String mapKey = "";
		if (ipBean.getTflag().equals(IpCheckConstants.IP_SOURCE_TYPE)) {
			mapKey = ipBean.getStorageType() + "_" + ipBean.getTflag() + "_" + ipBean.getBbdw_id() + "_"
					+ ipBean.getLyid();
		} else if (ipBean.getTflag().equals(IpCheckConstants.IP_ALLOCATE_TYPE)) {
			if (ipBean.getFpfs().equals(IpAllocateType.RE_ALLOCATE.getTypeValue())) {
				mapKey = ipBean.getStorageType() + "_" + ipBean.getTflag() + "_" + ipBean.getBbdw_id() + "_"
						+ ipBean.getFpfs() + "_" + ipBean.getFpdx();
			} else {
				mapKey = ipBean.getStorageType() + "_" + ipBean.getTflag() + "_" + ipBean.getBbdw_id() + "_"
						+ ipBean.getFpfs();
			}
		}
		if (set.contains(mapKey)) {
			flag = false;
		} else {
			set.add(mapKey);
		}

		return flag;
	}

}
